package bigdata.hermesfuxi.eagle.etl.jobs;

import bigdata.hermesfuxi.eagle.etl.bean.DataLogBean;
import com.alibaba.fastjson.JSON;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;

import java.util.HashMap;
import java.util.Map;
import java.util.Properties;
import java.util.Random;

/**
 * 测试数据生成器
 */
public class TestDataGenerator {
    public static void main(String[] args) {
        Properties props = new Properties();
        props.put("bootstrap.servers", "localhost:9092");
        props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        
        KafkaProducer<String, String> producer = new KafkaProducer<>(props);
        Random random = new Random();
        
        String[] anchorIds = {"anchor001", "anchor002", "anchor003", "anchor004", "anchor005"};
        Integer[] giftIds = {1, 2, 3, 4, 5};
        String[] giftNames = {"玫瑰", "跑车", "飞机", "火箭", "城堡"};
        
        try {
            for (int i = 0; i < 100; i++) {
                DataLogBean dataLogBean = new DataLogBean();
                dataLogBean.setEventId("liveReward");
                dataLogBean.setDeviceId("device" + random.nextInt(1000));
                dataLogBean.setTimestamp(System.currentTimeMillis());
                
                Map<String, Object> properties = new HashMap<>();
                properties.put("anchor_id", anchorIds[random.nextInt(anchorIds.length)]);
                properties.put("gift_id", giftIds[random.nextInt(giftIds.length)]);
                dataLogBean.setProperties(properties);
                
                String json = JSON.toJSONString(dataLogBean);
                ProducerRecord<String, String> record = new ProducerRecord<>("live-gift-events", json);
                producer.send(record);
                
                System.out.println("发送数据: " + json);
                Thread.sleep(1000);
            }
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            producer.close();
        }
    }
}
